package com.zcy;

import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.runtime.client.EPEventService;
import com.zcy.event.MyEvent;
import com.zcy.util.EsperBuilder;
import com.zcy.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * 14.3.6. Externally-timed Batch Window (ext_timed_batch or win:ext_timed_batch)
 * 称为滚动窗口（Tumbling Window)
 * Ext-timed Batch Window是Esper中最像FLink中的TimeWindow的一种窗口，是一种真正由时间时间进行驱动的窗口
 */
@Slf4j
public class ExtTimeBatchWindow extends TimeUtils {

    private static final String _TimeBatch = "@name('extTime-batch') select sum(num) from MyEvent#win:ext_timed_batch(timestamp, 2 sec)";
    private static final String _TimeBatchWIthDelay = "@name('extTime-batch') select sum(num) from MyEvent#win:ext_timed_batch(timestamp, 2 sec, 3000L)";

    /**
     * 将外部定时批处理窗口与常规时间批处理窗口进行比较的主要区别在于，该窗口的滚动不是基于运行时间，而是严格根据时间戳记表达式的结果（当针对进入窗口的事件进行评估时）。
     * <p>
     * 窗口底层的算法将最旧事件到达时表达式返回的时间戳值与事件到达时最新到达事件的表达式返回的时间戳值进行比较。
     * 如果时间戳值之间的时间间隔大于计时器周期参数，则该算法将发布当前的事件批次。
     * 因此，该窗口仅在事件到达时才发布批处理，并且仅考虑每个事件的timestamp属性（或返回的其他表达式值），而不考虑运行时间。
     * sum is : 15
     * sum is : 20
     * sum is : 30
     */
    @Test
    public void extBatchWindowTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_TimeBatch, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> Stream.of(eventBeans).forEach(item -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        }));

        epEventService.sendEventBean(MyEvent.builder().timestamp(0).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        epEventService.sendEventBean(MyEvent.builder().timestamp(1000).amount("001").name("001-name").num(5).build(), MyEvent.class.getSimpleName());
        epEventService.sendEventBean(MyEvent.builder().timestamp(2000).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("003").name("003-name").num(30).build(), MyEvent.class.getSimpleName());
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("004").name("004-name").num(40).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }

    /**
     * sum is : 10
     * sum is : 25
     * sum is : 30
     */
    @Test
    public void extBatchWindowWithDelayTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_TimeBatchWIthDelay, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> Stream.of(eventBeans).forEach(item -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        }));

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 1L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(5).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 1L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 3L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("003").name("003-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 4L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("004").name("004-name").num(40).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }


}
